{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(1)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "pd.set_option('display.max_colwidth', 1000)\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"temp\" + os.sep\n",
    "LOCAL_DIR = DATA_DIR\n",
    "\n",
    "OSS_END_POINT = \"*\";\n",
    "OSS_BUCKET_NAME = \"*\";\n",
    "OSS_ACCESS_ID = \"*\";\n",
    "OSS_ACCESS_KEY = \"*\";\n",
    "\n",
    "OSS_PREFIX_URI = \"oss://\" + OSS_BUCKET_NAME + \"/\";\n",
    "\n",
    "HDFS_URI = \"hdfs://10.*.*.*:9000/\";\n",
    "\n",
    "IRIS_HTTP_URL = \"http://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data\";\n",
    "\n",
    "IRIS_SCHEMA_STR = \"sepal_length double, sepal_width double, petal_length double, petal_width double, category string\";"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_1\n",
    "import datetime\n",
    "\n",
    "local = LocalFileSystem();\n",
    "print(local.getHomeDirectory());\n",
    "print(local.getKind());\n",
    "\n",
    "if not local.exists(LOCAL_DIR) :\n",
    "    local.mkdirs(LOCAL_DIR);\n",
    "\n",
    "for status in local.listStatus(LOCAL_DIR) :\n",
    "    print(str(status.getPath().toUri()) + \" \\t\" + str(status.getLen()) + \" \\t\" \n",
    "          + str(datetime.datetime.utcfromtimestamp(status.getModificationTime()/1000)))\n",
    "\n",
    "path = LOCAL_DIR + \"hello.txt\";\n",
    "\n",
    "fos = local.create(path, True)\n",
    "fos.write(b'Hello Alink!')\n",
    "fos.flush()\n",
    "fos.close()\n",
    "\n",
    "status = local.getFileStatus(path)\n",
    "print(status)\n",
    "print(status.getLen())\n",
    "print(datetime.datetime.utcfromtimestamp(status.getModificationTime()/1000));\n",
    "\n",
    "nBytes = local.getFileStatus(path).getLen()\n",
    "fis = local.open(path)\n",
    "(numBytesRead, b) = fis.read(nBytes, 0)\n",
    "fis.close()\n",
    "print(b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_2_1\n",
    "hdfs = HadoopFileSystem(HDFS_URI);\n",
    "print(hdfs.getKind());\n",
    "\n",
    "hdfsDir = HDFS_URI + \"user/yangxu/alink/data/temp/\";\n",
    "\n",
    "if not hdfs.exists(hdfsDir) :\n",
    "    hdfs.mkdirs(hdfsDir);\n",
    "\n",
    "for status in hdfs.listStatus(hdfsDir) :\n",
    "    print(str(status.getPath().toUri()) + \" \\t\" + str(status.getLen()) + \" \\t\" \n",
    "          + str(datetime.datetime.utcfromtimestamp(status.getModificationTime()/1000)))\n",
    "\n",
    "path = hdfsDir + \"hello.txt\";\n",
    "\n",
    "fos = hdfs.create(path, True)\n",
    "fos.write(b'Hello Alink!')\n",
    "fos.flush()\n",
    "fos.close()\n",
    "\n",
    "nBytes = hdfs.getFileStatus(path).getLen()\n",
    "fis = hdfs.open(path)\n",
    "(numBytesRead, bytesRead) = fis.read(nBytes, 0)\n",
    "print(bytesRead)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def copy(fs_from, path_from, fs_to, path_to) :\n",
    "    nBytes = fs_from.getFileStatus(path_from).getLen()\n",
    "    fis = fs_from.open(path_from)\n",
    "    (numBytesRead, bytesRead) = fis.read(nBytes, 0)\n",
    "    fis.close()\n",
    "    fos = fs_to.create(path_to, True)\n",
    "    fos.write(bytesRead)\n",
    "    fos.flush()\n",
    "    fos.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_2_2\n",
    "local = LocalFileSystem();\n",
    "\n",
    "hdfs = HadoopFileSystem(HDFS_URI);\n",
    "\n",
    "copy(hdfs, HDFS_URI + \"user/yangxu/alink/data/temp/hello.txt\",\n",
    "     local, LOCAL_DIR + \"hello_1.txt\");\n",
    "\n",
    "copy(local, LOCAL_DIR + \"hello_1.txt\", \n",
    "     hdfs, HDFS_URI + \"user/yangxu/alink/data/temp/hello_2.txt\")\n",
    "\n",
    "for status in hdfs.listStatus(HDFS_URI + \"user/yangxu/alink/data/temp/\") :\n",
    "    print(str(status.getPath().toUri()) + \" \\t\" + str(status.getLen()) + \" \\t\" \n",
    "          + str(datetime.datetime.utcfromtimestamp(status.getModificationTime()/1000)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_3_1\n",
    "oss = OssFileSystem(OSS_END_POINT,OSS_BUCKET_NAME,OSS_ACCESS_ID,OSS_ACCESS_KEY);\n",
    "\n",
    "print(oss.getKind());\n",
    "\n",
    "ossDir = OSS_PREFIX_URI + \"alink/data/temp/\";\n",
    "\n",
    "if not oss.exists(ossDir) :\n",
    "    oss.mkdirs(ossDir);\n",
    "\n",
    "for status in oss.listStatus(ossDir) :\n",
    "    print(str(status.getPath().toUri()) + \" \\t\" + str(status.getLen()) + \" \\t\" \n",
    "          + str(datetime.datetime.utcfromtimestamp(status.getModificationTime()/1000)))\n",
    "\n",
    "path = ossDir + \"hello.txt\";\n",
    "\n",
    "fos = oss.create(path, True)\n",
    "fos.write(b'Hello Alink!')\n",
    "fos.flush()\n",
    "fos.close()\n",
    "\n",
    "nBytes = oss.getFileStatus(path).getLen()\n",
    "fis = oss.open(path)\n",
    "(numBytesRead, bytesRead) = fis.read(nBytes, 0)\n",
    "print(bytesRead)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_3_2\n",
    "\n",
    "local = LocalFileSystem();\n",
    "\n",
    "oss = OssFileSystem(OSS_END_POINT,OSS_BUCKET_NAME,OSS_ACCESS_ID,OSS_ACCESS_KEY);\n",
    "\n",
    "copy(oss, OSS_PREFIX_URI + \"alink/data/temp/hello.txt\",\n",
    "     local, LOCAL_DIR + \"hello_1.txt\");\n",
    "\n",
    "copy(local, LOCAL_DIR + \"hello_1.txt\",\n",
    "     oss, OSS_PREFIX_URI + \"alink/data/temp/hello_2.txt\");\n",
    "\n",
    "for status in hdfs.listStatus(HDFS_URI + \"user/yangxu/alink/data/temp/\") :\n",
    "    print(str(status.getPath().toUri()) + \" \\t\" + str(status.getLen()) + \" \\t\" \n",
    "          + str(datetime.datetime.utcfromtimestamp(status.getModificationTime()/1000)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "help(CsvSourceBatchOp.setFilePath)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_1_1\n",
    "source_local = CsvSourceBatchOp()\\\n",
    ".setFilePath(LOCAL_DIR + \"iris.data\")\\\n",
    ".setSchemaStr(\"sepal_length double, sepal_width double, \"\n",
    "              + \"petal_length double, petal_width double, category string\");\n",
    "\n",
    "source_local.firstN(5).print();\n",
    "\n",
    "source_url = CsvSourceBatchOp()\\\n",
    ".setFilePath(\"http://archive.ics.uci.edu/ml/machine-learning-databases\"\n",
    "             + \"/iris/iris.data\")\\\n",
    ".setSchemaStr(\"sepal_length double, sepal_width double, \"\n",
    "              + \"petal_length double, petal_width double, category string\");\n",
    "\n",
    "source_url.firstN(5).print();\n",
    "\n",
    "source_stream = CsvSourceStreamOp()\\\n",
    ".setFilePath(\"http://archive.ics.uci.edu/ml/machine-learning-databases\"\n",
    "             + \"/iris/iris.data\")\\\n",
    ".setSchemaStr(\"sepal_length double, sepal_width double, \"\n",
    "              + \"petal_length double, petal_width double, category string\");\n",
    "\n",
    "source_stream.filter(\"sepal_length < 4.5\").print();\n",
    "StreamOperator.execute();\n",
    "\n",
    "wine_url = CsvSourceBatchOp()\\\n",
    ".setFilePath(\"http://archive.ics.uci.edu/ml/machine-learning-databases\"\n",
    "             + \"/wine-quality/winequality-white.csv\")\\\n",
    ".setSchemaStr(\"fixedAcidity double,volatileAcidity double,citricAcid double,\"\n",
    "              + \"residualSugar double, chlorides double,freeSulfurDioxide double,\"\n",
    "              + \"totalSulfurDioxide double,density double, pH double,\"\n",
    "              + \"sulphates double,alcohol double,quality double\")\\\n",
    ".setFieldDelimiter(\";\")\\\n",
    ".setIgnoreFirstLine(True);\n",
    "\n",
    "wine_url.firstN(5).print();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_1_2\n",
    "oss = OssFileSystem(OSS_END_POINT,OSS_BUCKET_NAME,OSS_ACCESS_ID,OSS_ACCESS_KEY);\n",
    "\n",
    "filePaths = [FilePath(LOCAL_DIR + \"iris.csv\"), \n",
    "             FilePath(HDFS_URI + \"user/yangxu/alink/data/temp/iris.csv\"),\n",
    "             FilePath(OSS_PREFIX_URI + \"alink/data/temp/iris.csv\", oss)]\n",
    "\n",
    "for filePath in filePaths :\n",
    "    print(filePath.getPathStr())\n",
    "    CsvSourceBatchOp()\\\n",
    "        .setFilePath(IRIS_HTTP_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .link(\n",
    "            CsvSinkBatchOp().setFilePath(filePath).setOverwriteSink(True)\n",
    "        );\n",
    "    BatchOperator.execute();\n",
    "\n",
    "    CsvSourceBatchOp()\\\n",
    "        .setFilePath(filePath)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .firstN(3)\\\n",
    "        .print()\n",
    "\n",
    "for filePath in filePaths :\n",
    "    print(filePath.getPathStr())\n",
    "    CsvSourceStreamOp()\\\n",
    "        .setFilePath(IRIS_HTTP_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .link(\n",
    "            CsvSinkStreamOp().setFilePath(filePath).setOverwriteSink(True)\n",
    "        );\n",
    "    StreamOperator.execute();\n",
    "\n",
    "    CsvSourceStreamOp()\\\n",
    "        .setFilePath(filePath)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .filter(\"sepal_length < 4.5\")\\\n",
    "        .print()\n",
    "    StreamOperator.execute();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_2\n",
    "\n",
    "TsvSourceBatchOp()\\\n",
    ".setFilePath(\"http://files.grouplens.org/datasets/movielens/ml-100k/u.data\")\\\n",
    ".setSchemaStr(\"user_id long, item_id long, rating float, ts long\")\\\n",
    ".firstN(5)\\\n",
    ".print();\n",
    "\n",
    "TextSourceBatchOp()\\\n",
    ".setFilePath(LOCAL_DIR + \"iris.scale\")\\\n",
    ".firstN(5)\\\n",
    ".print();\n",
    "\n",
    "LibSvmSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"iris.scale\")\\\n",
    "    .firstN(5)\\\n",
    "    .lazyPrint(-1, \"< read by LibSvmSourceBatchOp >\")\\\n",
    "    .link(\n",
    "        VectorNormalizeBatchOp().setSelectedCol(\"features\")\n",
    "    )\\\n",
    "    .lazyPrint(-1, \"< after VectorNormalize >\")\n",
    "BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_3_1\n",
    "oss = OssFileSystem(OSS_END_POINT,OSS_BUCKET_NAME,OSS_ACCESS_ID,OSS_ACCESS_KEY);\n",
    "\n",
    "filePaths = [FilePath(LOCAL_DIR + \"iris.ak\"), \n",
    "             FilePath(HDFS_URI + \"user/yangxu/alink/data/temp/iris.ak\"),\n",
    "             FilePath(OSS_PREFIX_URI + \"alink/data/temp/iris.ak\", oss)]\n",
    "\n",
    "for filePath in filePaths :\n",
    "    print(filePath.getPathStr())\n",
    "    CsvSourceBatchOp()\\\n",
    "        .setFilePath(IRIS_HTTP_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .link(\n",
    "            AkSinkBatchOp().setFilePath(filePath).setOverwriteSink(True)\n",
    "        );\n",
    "    BatchOperator.execute();\n",
    "\n",
    "    AkSourceBatchOp()\\\n",
    "        .setFilePath(filePath)\\\n",
    "        .firstN(3)\\\n",
    "        .print()\n",
    "\n",
    "for filePath in filePaths :\n",
    "    print(filePath.getPathStr())\n",
    "    CsvSourceStreamOp()\\\n",
    "        .setFilePath(IRIS_HTTP_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .link(\n",
    "            AkSinkStreamOp().setFilePath(filePath).setOverwriteSink(True)\n",
    "        );\n",
    "    StreamOperator.execute();\n",
    "\n",
    "    AkSourceStreamOp()\\\n",
    "        .setFilePath(filePath)\\\n",
    "        .filter(\"sepal_length < 4.5\")\\\n",
    "        .print()\n",
    "    StreamOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_4\n",
    "ParquetSourceBatchOp()\\\n",
    "    .setFilePath(\"https://alink-test-data.oss-cn-hangzhou.aliyuncs.com/iris.parquet\")\\\n",
    "    .lazyPrintStatistics()\\\n",
    "    .print()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ParquetSourceStreamOp()\\\n",
    "    .setFilePath(\"https://alink-test-data.oss-cn-hangzhou.aliyuncs.com/iris.parquet\")\\\n",
    "    .print()\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_5_1\n",
    "import datetime\n",
    "\n",
    "@udf(input_types=[AlinkDataTypes.BIGINT()], result_type=AlinkDataTypes.TIMESTAMP(3)) \n",
    "def from_unix_timestamp(ts):\n",
    "    return datetime.datetime.fromtimestamp(ts)\n",
    "\n",
    "source = TsvSourceStreamOp()\\\n",
    "    .setFilePath(\"http://files.grouplens.org/datasets/movielens/ml-100k/u.data\")\\\n",
    "    .setSchemaStr(\"user_id long, item_id long, rating float, ts long\")\\\n",
    "    .link(\n",
    "        UDFStreamOp()\\\n",
    "            .setFunc(from_unix_timestamp)\\\n",
    "            .setSelectedCols([\"ts\"])\\\n",
    "            .setOutputCol(\"ts\")\n",
    "    )\n",
    "\n",
    "source.link(\n",
    "    Export2FileSinkStreamOp()\\\n",
    "        .setFilePath(LOCAL_DIR + \"with_local_time\")\\\n",
    "        .setWindowTime(5)\\\n",
    "        .setOverwriteSink(True)\n",
    ")\n",
    "\n",
    "source.link(\n",
    "    AkSinkStreamOp()\\\n",
    "        .setFilePath(LOCAL_DIR + \"ratings.ak\")\\\n",
    "        .setOverwriteSink(True)\n",
    ")\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"with_local_time\")\\\n",
    "    .lazyPrintStatistics(\"Statistics for data in the folder 'with_local_time' : \")\n",
    "BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_5_2\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"ratings.ak\")\\\n",
    "    .orderBy(\"ts\", 1000000)\\\n",
    "    .lazyPrintStatistics(\"Statistics for data in the file 'ratings.ak' : \")\\\n",
    "    .link(\n",
    "        AkSinkBatchOp()\\\n",
    "            .setFilePath(LOCAL_DIR + \"ratings_ordered.ak\")\\\n",
    "            .setOverwriteSink(True)\n",
    "    )\n",
    "BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"ratings_ordered.ak\")\\\n",
    "    .link(\n",
    "        Export2FileSinkStreamOp()\\\n",
    "            .setFilePath(LOCAL_DIR + \"with_ts_time\")\\\n",
    "            .setTimeCol(\"ts\")\\\n",
    "            .setWindowTime(3600 * 24)\\\n",
    "            .setOverwriteSink(True)\n",
    "    )\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"with_ts_time\")\\\n",
    "    .lazyPrintStatistics(\"Statistics for data in the folder 'with_ts_time' : \")\n",
    "\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"with_ts_time\" + os.sep + \"199709210000000\")\\\n",
    "    .print()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_5_3\n",
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"ratings_ordered.ak\")\\\n",
    "    .link(\n",
    "        Export2FileSinkStreamOp()\\\n",
    "            .setFilePath(LOCAL_DIR + \"data_with_partitions\")\\\n",
    "            .setTimeCol(\"ts\")\\\n",
    "            .setWindowTime(3600 * 24)\\\n",
    "            .setPartitionsFormat(\"year=yyyy/month=MM/day=dd\")\\\n",
    "            .setOverwriteSink(True)\n",
    "    )\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_6\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"data_with_partitions\")\\\n",
    "    .lazyPrintStatistics(\"Statistics for data in the folder 'data_with_partitions' : \")\n",
    "\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"data_with_partitions\")\\\n",
    "    .setPartitions(\"year='1997'\")\\\n",
    "    .lazyPrintStatistics(\"Statistics for data of year=1997 : \")\n",
    "\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"data_with_partitions\")\\\n",
    "    .setPartitions(\"year='1997' AND month>='10'\")\\\n",
    "    .lazyPrintStatistics(\"Statistics for data of year 1997's last 3 months : \")\n",
    "\n",
    "BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"data_with_partitions\")\\\n",
    "    .setPartitions(\"day LIKE '3_'\")\\\n",
    "    .lazyPrint(10, \">>> day LIKE '3_'\")\n",
    "\n",
    "AkSourceBatchOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"data_with_partitions\")\\\n",
    "    .setPartitions(\"day LIKE '%3%'\")\\\n",
    "    .print(10, \">>> day LIKE '%3%'\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(LOCAL_DIR + \"data_with_partitions\")\\\n",
    "    .setPartitions(\"day IN('01', '02')\")\\\n",
    "    .sample(0.001)\\\n",
    "    .print()\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
